package com.shuiyun.canal.task;

import cn.hutool.core.date.DateUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.shuiyun.canal.entity.vo.ColumnValuePair;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @program: base-canal
 * @description: 定时读取binlog日志
 * @author: shuiyun
 * @create: 2020-07-27 11:06
 **/
@RestController
@RequestMapping("/task")
@Slf4j
public class BinLogRead implements ApplicationRunner {

    @Value("${canal.host}")
    private String host;
    @Value("${canal.port}")
    private int port;
    @Value("${canal.username}")
    private String username;
    @Value("${canal.password}")
    private String password;
    @Value("${canal.instance}")
    private String instance;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        CanalConnector conn = getConn();
        while (true){
            conn.connect();
            //订阅实例中所有的数据库和表
//            conn.subscribe("used_car_dev\\.asset.*,used_car_dev\\.organization.*");
            conn.subscribe("GPSKernel.GPSVehicle");
//            conn.subscribe();
            // 回滚到未进行ack的地方
            conn.rollback();
            // 获取数据 每次获取一百条改变数据
            Message message = conn.getWithoutAck(999999999,1L,TimeUnit.SECONDS);

            long id = message.getId();
            int size = message.getEntries().size();
            if (id != -1 && size >0){
                // 数据解析
                analysis(message.getEntries());
            }
            // 确认消息
            conn.ack(message.getId());
            // 关闭连接
            conn.disconnect();
        }
    }

    /**
     * 数据解析
     * @param
     * @Author: shuiyun
     * @Date: 2020/7/27 11:28
     * @return
     */
    private void analysis(List<CanalEntry.Entry> entries) {
        String transId = null;
        for (CanalEntry.Entry entry : entries) {
            // 只解析mysql事务的操作，其他的不解析
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ) {
                transId = UUID.randomUUID().toString();
                continue;
            }
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND){
                transId = null;
                continue;
            }

            // 解析binlog
            CanalEntry.RowChange rowChange = null;

            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 获取操作类型字段（增加  删除  修改）
            CanalEntry.EventType eventType = rowChange.getEventType();
            // 获取当前操作所属的数据库
            String dbName = entry.getHeader().getSchemaName();
            // 获取当前操作所属的表
            String tableName = entry.getHeader().getTableName();
            // 事务提交时间
            Long timestamp = entry.getHeader().getExecuteTime();
//            log.info(DateUtil.formatDateTime(new Date(timestamp)));
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                dataDetails(rowData.getBeforeColumnsList(),rowData.getAfterColumnsList(), dbName, tableName, eventType,timestamp,transId);
                log.info("-------------------------------------------------------------");
            }
        }
    }



    /**
     * 解析具体一条Binlog消息的数据
     * @param dbName        当前操作所属数据库名称
     * @param tableName     当前操作所属表名称
     * @param eventType     当前操作类型（新增、修改、删除）
     */
    private static void dataDetails(List<CanalEntry.Column> beforeColumns,
                                    List<CanalEntry.Column> afterColumns,
                                    String dbName,
                                    String tableName,
                                    CanalEntry.EventType eventType,
                                    long timestamp,String transId) {
        if (CanalEntry.EventType.QUERY.equals(eventType)){
            return;
        }

        // 找到当前那些列发生了改变  以及改变的值
        List<ColumnValuePair> beforeColumnValueList = new ArrayList<ColumnValuePair>();
        List<ColumnValuePair> afterColumnValueList = new ArrayList<ColumnValuePair>();

        for (CanalEntry.Column column : beforeColumns) {
            ColumnValuePair columnValuePair = new ColumnValuePair(column.getName(), column.getValue(), column.getUpdated());
            beforeColumnValueList.add(columnValuePair);
        }
        for (CanalEntry.Column column : afterColumns) {
            ColumnValuePair columnValuePair = new ColumnValuePair(column.getName(), column.getValue(), column.getUpdated());
            afterColumnValueList.add(columnValuePair);
        }

        String key = UUID.randomUUID().toString();


        log.info("数据库："+ dbName);
        log.info("表名："+ tableName);
        log.info("操作类型:"+ eventType);
        int size = afterColumns.size();
        if (CanalEntry.EventType.INSERT.equals(eventType)){
            log.info("新增数据："+afterColumnValueList);
        }else if(CanalEntry.EventType.DELETE.equals(eventType)){
            log.info("删除数据："+beforeColumnValueList);
        }else{
            log.info("更新数据：");
            for (int i=0;i< size;i++){
                ColumnValuePair after = afterColumnValueList.get(i);
                ColumnValuePair before = null;
                try {
                    before = beforeColumnValueList.get(i);
                }catch (Exception e){

                }
                if (after.getIsValid()){
                    String beforeValue = null == before?"":before.getColumnValue();
                    log.info("列"+after.getColumnName()+"发生改变，改变之前"+beforeValue+",改变之后："+after.getColumnValue());
                }
            }
        }
        log.info("事务ID："+transId);
        log.info("操作时间："+ DateUtil.formatDateTime(new Date(timestamp)));

    }



    /**
     * 获取连接
     * @param
     * @Author: shuiyun
     * @Date: 2020/7/23 14:57
     * @return
     */
    public CanalConnector getConn() {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), instance, username, password);
        return canalConnector;
    }


}
